package com.learning.rocketmq.springbootclient.springboot;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.rocketmq.starter.core.producer.MessageProxy;
import org.rocketmq.starter.core.producer.RocketMQProducerTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.Charset;
import java.util.List;

/**
 * ClassName: ProducerServiceBoot
 * Description:
 * Date: 2019/1/15 13:51 【需求编号】
 *
 * @author Sam Sho
 * @version V1.0.0
 */
@Service
public class ProducerServiceBoot {

    @Resource
    private RocketMQProducerTemplate producerTemplate;

    /**
     * 简单使用
     *
     * @throws InterruptedException
     * @throws RemotingException
     * @throws MQClientException
     * @throws MQBrokerException
     */
    public void send() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message("topic", "msg context".getBytes(Charset.forName("utf-8")));
        producerTemplate.send(message);
        producerTemplate.start();
    }

    /**
     * MessageQueueSelector
     *
     * @throws Exception
     */
    public void send2() throws Exception {
        Message message = new Message("topic", "msg context".getBytes(Charset.forName("utf-8")));

        MessageProxy messageProxy = new MessageProxy();
        messageProxy.setMessage(message);
        messageProxy.setMessageQueueSelector(new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                return list.get(0);
            }
        });
        producerTemplate.send(messageProxy);
        producerTemplate.start();
    }


    /**
     * 异步
     *
     * @throws Exception
     */
    public void send3() throws Exception {
        Message message = new Message("topic", "msg context".getBytes(Charset.forName("utf-8")));

        MessageProxy messageProxy = new MessageProxy();
        messageProxy.setSendCallback(new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                SendStatus sendStatus = sendResult.getSendStatus();
                switch (sendStatus) {
                    case SEND_OK:
                        break;
                    case FLUSH_DISK_TIMEOUT:
                        break;
                    case FLUSH_SLAVE_TIMEOUT:
                        break;
                    case SLAVE_NOT_AVAILABLE:
                        break;
                }
            }

            @Override
            public void onException(Throwable e) {

            }
        });
        producerTemplate.send(messageProxy);
        producerTemplate.start();
    }


}
